package com.atguigu.tingshu.search.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.extra.pinyin.PinyinUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggestOption;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.AlbumFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.model.album.*;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.Repository.AlbumInfoIndexRepository;
import com.atguigu.tingshu.search.Repository.SuggestIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserFeignClient;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"all"})
public class SearchServiceImpl implements SearchService {

    @Autowired
    private AlbumFeignClient albumFeignClient;

    @Autowired
    private AlbumInfoIndexRepository albumInfoIndexRepository;

    @Autowired
    private UserFeignClient userFeignClient;

    @Autowired
    @Qualifier("threadPoolTaskExecutor")
    private Executor threadPoolTaskExecutor;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private SuggestIndexRepository suggestIndexRepository;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;


    @Override
    public void upperAlbum(Long albumId) {
        //1.创建上传信息
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();

        CompletableFuture<AlbumInfo> albumInfoIndexCompletableFuture = CompletableFuture.supplyAsync(() -> {
            //2.调用远程接口获取专辑信息
            AlbumInfo albumInfo = albumFeignClient.getAlbumInfo(albumId).getData();
            Assert.notNull(albumInfo, "专辑不存在");
            BeanUtil.copyProperties(albumInfo, albumInfoIndex);
            List<AlbumAttributeValue> albumAttributeValueVoList = albumInfo.getAlbumAttributeValueVoList();
            if (CollUtil.isNotEmpty(albumAttributeValueVoList)) {
                List<AttributeValueIndex> attributeValueIndexList = albumAttributeValueVoList
                        .stream()
                        .map(aav -> BeanUtil.copyProperties(aav, AttributeValueIndex.class))
                        .collect(Collectors.toList());
                albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
            }
            return albumInfo;
        }, threadPoolTaskExecutor);
        //设置四个属性的参数 在专辑中获取使用stream流的方式进行获取

        //3.调用远程接口获取专辑分类
        CompletableFuture<Void> completableFuture = albumInfoIndexCompletableFuture.thenAcceptAsync(albumInfo -> {
            Long category3Id = albumInfo.getCategory3Id();
            BaseCategoryView baseCategoryView = albumFeignClient.getCategoryView(category3Id).getData();
            Assert.notNull(baseCategoryView, "专辑分类不存在");
            albumInfoIndex.setCategory1Id(baseCategoryView.getCategory1Id());
            albumInfoIndex.setCategory2Id(baseCategoryView.getCategory2Id());
        }, threadPoolTaskExecutor);

        //4.获取用户的姓名

        CompletableFuture<Void> thenAcceptAsync = albumInfoIndexCompletableFuture.thenAcceptAsync(albumInfo -> {
            UserInfoVo userInfoVo = userFeignClient.getUserInfoVo(albumInfo.getUserId()).getData();
            Assert.notNull(userInfoVo, "用户不存在");
            albumInfoIndex.setAnnouncerName(userInfoVo.getNickname());
        }, threadPoolTaskExecutor);

        CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {
            int num1 = RandomUtil.randomInt(3000);
            int num2 = RandomUtil.randomInt(2000);
            int num3 = RandomUtil.randomInt(1000);
            int num4 = RandomUtil.randomInt(500);
            albumInfoIndex.setPlayStatNum(num1);
            albumInfoIndex.setSubscribeStatNum(num2);
            albumInfoIndex.setBuyStatNum(num3);
            albumInfoIndex.setCommentStatNum(num4);
            BigDecimal score = BigDecimal.valueOf(num1).multiply(BigDecimal.valueOf(0.1))
                    .add(BigDecimal.valueOf(num2).multiply(BigDecimal.valueOf(0.2)))
                    .add(BigDecimal.valueOf(num3).multiply(BigDecimal.valueOf(0.3)))
                    .add(BigDecimal.valueOf(num4).multiply(BigDecimal.valueOf(0.4)));
            albumInfoIndex.setHotScore(score.doubleValue());
        }, threadPoolTaskExecutor);
        CompletableFuture.allOf(albumInfoIndexCompletableFuture,
                        thenAcceptAsync,
                        completableFuture,
                        completableFuture1).orTimeout(1, TimeUnit.SECONDS)
                .join();

        albumInfoIndexRepository.save(albumInfoIndex);
        this.saveSuggestIndex(albumInfoIndex);

        RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        bloomFilter.add(albumId);
    }

    //todo 复制的方法

    /**
     * 将专辑标题存入提示词索引库，用于自动补全
     *
     * @param albumTitle
     */
    private void saveSuggestIndex(AlbumInfoIndex albumInfoIndex) {
        SuggestIndex suggestIndex = new SuggestIndex();
        suggestIndex.setId(albumInfoIndex.getId().toString());
        String albumTitle = albumInfoIndex.getAlbumTitle();
        suggestIndex.setTitle(albumTitle);
        suggestIndex.setKeyword(new Completion(new String[]{albumTitle}));
        String pinyin = PinyinUtil.getPinyin(albumTitle, "");
        suggestIndex.setKeywordPinyin(new Completion(new String[]{pinyin}));
        String firstLetter = PinyinUtil.getFirstLetter(albumTitle, "");
        suggestIndex.setKeywordSequence(new Completion(new String[]{firstLetter}));
        suggestIndexRepository.save(suggestIndex);
    }

    @Override
    public void lowerAlbum(Long albumId) {
        albumInfoIndexRepository.deleteById(albumId);
        log.info("下架专辑{}成功", albumId);
        suggestIndexRepository.deleteById(albumId.toString());
    }

    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery query) {

        try {
            //1.设置查询条件
            SearchRequest searchRequest = this.buildDSL(query);
            System.err.println(searchRequest);
            //2.执行条件查询
            SearchResponse<AlbumInfoIndex> search = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);
            //3.返回结果
            AlbumSearchResponseVo albumSearchResponseVo = this.parseResult(query, search);
            return albumSearchResponseVo;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    private static final String INDEX_NAME = "albuminfo";

    @Override
    public SearchRequest buildDSL(AlbumIndexQuery query) {
        SearchRequest.Builder builder = new SearchRequest.Builder();
        //设置关键字查询
        builder.index(INDEX_NAME);
        String keyword = query.getKeyword();
        //2.设置查询条件，请求体中“query” 三大条件选择创建一个bool查询对象组合条件
        BoolQuery.Builder allConditionBoolQueryBuilder = new BoolQuery.Builder();

        if (StringUtils.isNotEmpty(keyword)) {
            allConditionBoolQueryBuilder.must(m -> m.match(m1 -> m1.field("albumTitle").query(keyword)));
        }
        //根据三个标题查询
        Long category1Id = query.getCategory1Id();
        Long category2Id = query.getCategory2Id();
        Long category3Id = query.getCategory3Id();
        if (category1Id != null) {
            allConditionBoolQueryBuilder.filter(f -> f.term(t -> t.field("category1Id").value(category1Id)));
        }
        if (category2Id != null) {
            allConditionBoolQueryBuilder.filter(f -> f.term(t -> t.field("category2Id").value(category2Id)));
        }
        if (category3Id != null) {
            allConditionBoolQueryBuilder.filter(f -> f.term(t -> t.field("category3Id").value(category3Id)));
        }
        //根据标签进行查询 标签的类型是nested
        List<String> attributeList = query.getAttributeList();
        if (CollUtil.isNotEmpty(attributeList)) {
            for (String attributeValue : attributeList) {
                String[] split = attributeValue.split(":");
                allConditionBoolQueryBuilder.filter(
                        f -> f.nested(
                                n -> n.path("attributeValueIndexList").query(
                                        q -> q.bool
                                                (b -> b.must(
                                                                m -> m.term(t -> t.field("attributeValueIndexList.attributeId").value(split[0])))
                                                        .must(m -> m.term(t -> t.field("attributeValueIndexList.attributeValueId").value(split[1])))))));
            }
        }

        builder.query(allConditionBoolQueryBuilder.build()._toQuery());
        //设置高亮查询根据关键字
        builder.highlight(h -> h.fields("albumTitle", hl -> hl.preTags("<span style='color:red'>").postTags("</span>")));
        //设置分页查询根据请求头里面携带的分页参数
        Integer pageNo = query.getPageNo();
        Integer pageSize = query.getPageSize();
        builder.from((pageNo - 1) * pageSize);
        builder.size(pageSize);
        //设置排序的字段
        String order = query.getOrder();
        if (StringUtils.isNotEmpty(order)) {

            String srotFileld = "";
            String[] split = order.split(":");
            switch (split[0]) {
                case "1":
                    srotFileld = "hotScore";
                    break;
                case "2":
                    srotFileld = "playStatNum";
                    break;
                case "3":
                    srotFileld = "createTime";
                    break;
            }
            SortOrder sortOrder = split[1].equals("desc") ? SortOrder.Desc : SortOrder.Asc;
            String finalSrotFileld = srotFileld;
            builder.sort(s -> s.field(f -> f.field(finalSrotFileld).order(sortOrder)));
        }
        //设置返回的结果
        builder.source(s -> s.filter(f -> f.excludes(
                "attributeValueIndexList",
                "hotScore",
                "commentStatNum",
                "buyStatNum",
                "subscribeStatNum",
                "announcerName"
        )));
        return builder.build();
    }

    @Override
    public AlbumSearchResponseVo parseResult(AlbumIndexQuery query, SearchResponse<AlbumInfoIndex> searchResponse) {
        //1.创建一个对象进行封装
        AlbumSearchResponseVo albumAttributeValueVo = new AlbumSearchResponseVo();
        //2.从传过来的参数赋值
        List<Hit<AlbumInfoIndex>> hits = searchResponse.hits().hits();
        List<AlbumInfoIndexVo> infoIndexVos = null;
        if (CollUtil.isNotEmpty(hits)) {
            infoIndexVos = hits.stream().map(h -> {
                AlbumInfoIndex source = h.source();
                Map<String, List<String>> highlight = h.highlight();
                if (CollUtil.isNotEmpty(highlight)) {
                    source.setAlbumTitle(highlight.get("albumTitle").get(0));
                }
                return BeanUtil.copyProperties(source, AlbumInfoIndexVo.class);
            }).collect(Collectors.toList());
        }
        //3.向里面的属性赋值
        Integer pageNo = query.getPageNo();
        Integer pageSize = query.getPageSize();
        long value = searchResponse.hits().total().value();
        Long totalPages = value % pageSize == 0 ? value / pageSize : value / pageSize + 1;
        albumAttributeValueVo.setTotal(value);
        albumAttributeValueVo.setPageNo(pageNo);
        albumAttributeValueVo.setPageSize(pageSize);
        albumAttributeValueVo.setTotalPages(totalPages);
        albumAttributeValueVo.setList(infoIndexVos);
        //4.返回结果
        return albumAttributeValueVo;
    }


    //TODO 复制老师的代码 以后需要自己加点班

    /**
     * 查询1级分类下置顶7个三级分类热门专辑列表
     *
     * @param category1Id
     * @return Map<String, Object>置顶分类热门专辑Map对象 集合
     */
    @Override
    public List<Map<String, Object>> getChannel(Long category1Id) {
        try {
            //1.根据1级分类ID查询7个置顶分类集合
            List<BaseCategory3> baseCategory3List = albumFeignClient.findTopBaseCategory3(category1Id).getData();
            Assert.notNull(baseCategory3List, "没有查询到分类信息");

            //1.1 将得到3级分类列类型从BaseCategory3转为FieldValue类型
            List<FieldValue> fieldValueList = baseCategory3List.
                    stream()
                    .map(c3 -> FieldValue.of(c3.getId()))
                    .collect(Collectors.toList());


            //1.2 将三级分类集合转为Map<三级分类ID，三级分类对象> 解析结果时候方便获取三级分类对象
            Map<Long, BaseCategory3> baseCategory3Map = baseCategory3List
                    .stream()
                    .collect(Collectors.toMap(BaseCategory3::getId, c3 -> c3));

            //2.调用ES进行检索
            SearchResponse<AlbumInfoIndex> searchResponse = elasticsearchClient.search(s -> s.index(INDEX_NAME)
                            .size(0)
                            .query(q -> q.terms(t -> t.field("category3Id").terms(t1 -> t1.value(fieldValueList))))
                            .aggregations("c3_agg", a -> a.terms(t -> t.field("category3Id").size(7))
                                    .aggregations("top6_agg", a1 -> a1.topHits(t -> t.sort(s1 -> s1.field(f -> f.field("hotScore").order(SortOrder.Desc)))
                                            .source(s1 -> s1.filter(f -> f.excludes("attributeValueIndexList",
                                                    "hotScore",
                                                    "commentStatNum",
                                                    "buyStatNum",
                                                    "subscribeStatNum",
                                                    "announcerName")))
                                            .size(6)))
                            ),
                    AlbumInfoIndex.class);

            //3.解析ES检索聚合结果
            Map<String, Aggregate> aggregations = searchResponse.aggregations();
            //3.1 处理分类聚合结果
            List<LongTermsBucket> c3AggBuckets = aggregations.get("c3_agg").lterms().buckets().array();
            if (CollUtil.isNotEmpty(c3AggBuckets)) {
                //3.2 遍历LongTermsBucket集合每遍历一次就产出"置顶热门专辑分类Map"
                List<Map<String, Object>> list = c3AggBuckets.stream()
                        .map(c3AggBucket -> {
                            //3.2.1 构建置顶分类热门专辑Map对象
                            Map<String, Object> map = new HashMap<>();
                            //3.2.2 处理Map中分类对象
                            long category3Id = c3AggBucket.key();
                            map.put("baseCategory3", baseCategory3Map.get(category3Id));
                            //3.3 处理Map中专辑列表
                            //3.3.1 获取top6_agg聚合结果
                            List<AlbumInfoIndex> top6AlbumList = c3AggBucket.aggregations().get("top6_agg").topHits().hits().hits()
                                    .stream().map(hit -> {
                                        String albumInfoJsonStr = hit.source().toString();
                                        return JSON.parseObject(albumInfoJsonStr, AlbumInfoIndex.class);
                                    }).collect(Collectors.toList());
                            //3.3.2 封装Map中专辑列表
                            map.put("list", top6AlbumList);
                            return map;
                        }).collect(Collectors.toList());
                return list;
            }
            return null;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private static final String SUGGEST_INDEX = "suggestinfo";

    /**
     * 根据用户已录入字符进行自动补全
     *
     * @param keyword
     * @return
     */
    @Override
    public List<String> completeSuggest(String keyword) {
        try {
            //1.调用ES自动补全
            SearchResponse<SuggestIndex> searchResponse = elasticsearchClient.search(
                    s -> s.index(SUGGEST_INDEX)
                            .suggest(
                                    s1 -> s1.suggesters("letter-suggest", s2 -> s2.prefix(keyword).completion(c -> c.field("keywordSequence").size(10).skipDuplicates(true)))
                                            .suggesters("pinyin-suggest", s2 -> s2.prefix(keyword).completion(c -> c.field("keywordPinyin").size(10).skipDuplicates(true)))
                                            .suggesters("keyword-suggest", s2 -> s2.prefix(keyword).completion(c -> c.field("keyword").size(10).skipDuplicates(true)))
                            ),
                    SuggestIndex.class);
            //2.解析建议词结果,如果不同建议器结果存在相同的建议词，则只返回一个
            Set<String> hashSet = new HashSet<>();
            hashSet.addAll(this.parseSuggestResult("letter-suggest", searchResponse.suggest()));
            hashSet.addAll(this.parseSuggestResult("pinyin-suggest", searchResponse.suggest()));
            hashSet.addAll(this.parseSuggestResult("keyword-suggest", searchResponse.suggest()));

            //3.如果建议词结果数量小于10个，尝试采用全文检索补全到10个
            if (hashSet.size() < 10) {
                SearchResponse<AlbumInfoIndex> search = elasticsearchClient.search(
                        s -> s.index(INDEX_NAME)
                                .query(q -> q.match(m -> m.field("albumTitle").query(keyword)))
                                .size(10)
                                .source(s1 -> s1.filter(f -> f.includes("albumTitle"))),
                        AlbumInfoIndex.class
                );
                List<Hit<AlbumInfoIndex>> hitList = search.hits().hits();
                if (CollUtil.isNotEmpty(hitList)) {
                    for (Hit<AlbumInfoIndex> hit : hitList) {
                        String albumTitle = hit.source().getAlbumTitle();
                        hashSet.add(albumTitle);
                        if (hashSet.size() >= 10) {
                            break;
                        }
                    }
                }
            }
            //4.响应结果
            if (hashSet.size() > 10) {
                return new ArrayList<>(hashSet).subList(0, 10);
            }
            return new ArrayList<>(hashSet);
        } catch (IOException e) {
            log.error("自动补全失败：{}", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 根据热度更新排行榜
     */
    private static final String INDEX_ALBUMINFO_NAME = "albuminfo";

    @Override
    public void updateLatelyAlbumRanking() {
        try {
            //1. 调用远程服务查询专辑id
            //1.1 一级分类的标签
            List<BaseCategory1> baseCategory1s = albumFeignClient.findAllCategory1().getData();
            Assert.notEmpty(baseCategory1s, "一级分类为空");
            List<Long> category1IdList =
                    baseCategory1s.stream().map(BaseCategory1::getId).collect(Collectors.toList());


            //2.封装查询条件 查看es里面热门的专辑 按照一定顺序进行查询
            for (Long category1Id : category1IdList) {
                String condition[] = new String[]{"subscribeStatNum", "playStatNum", "buyStatNum", "commentStatNum", "hotScore"};
                for (String field : condition) {
                    SearchResponse<AlbumInfoIndex> searchResponse = elasticsearchClient
                            .search(s -> s.index(INDEX_ALBUMINFO_NAME)
                                    .query(q -> q.term
                                            (t -> t.field("category1Id").value(category1Id)))
                                    .sort(s1 -> s1.field(f -> f.field(field).order(SortOrder.Desc)))
                                    .size(10), AlbumInfoIndex.class);
                    System.err.println(searchResponse);
                    List<Hit<AlbumInfoIndex>> hitList = searchResponse.hits().hits();
                    //3. 将相应的结果储存到数据库里面
                    if (CollUtil.isNotEmpty(hitList)) {
                        List<AlbumInfoIndex> albumInfoIndexList = hitList.stream()
                                .map(Hit::source)
                                .collect(Collectors.toList());
                        String key = RedisConstant.RANKING_KEY_PREFIX + category1Id;
                        BoundHashOperations<String, String, List<AlbumInfoIndex>> boundHashOperations = redisTemplate.boundHashOps(key);
                        //3.1 进行保存操作
                        boundHashOperations.put(field, albumInfoIndexList);

                    }
                }
            }
        } catch (IOException e) {
            log.info("更新排行榜失败：{}", e);
            throw new RuntimeException(e);
        }
    }

    @Override
    public List<AlbumInfoIndex> findRankingList(Long category1Id, String dimension) {
        //1 获取redis的key
        String key = RedisConstant.RANKING_KEY_PREFIX + category1Id;
        //2 查询redis
        BoundHashOperations boundHashOperations = redisTemplate.boundHashOps(key);
        List<AlbumInfoIndex> albumInfoIndex = (List<AlbumInfoIndex>) boundHashOperations.get(dimension);
        return albumInfoIndex;
    }


    /**
     * 解析建议器结果
     *
     * @param suggestName
     * @param suggest
     * @return
     */
    private Collection<String> parseSuggestResult(String suggestName, Map<String, List<Suggestion<SuggestIndex>>> suggest) {
        List<String> list = new ArrayList<>();
        List<Suggestion<SuggestIndex>> suggestions = suggest.get(suggestName);
        if (CollUtil.isNotEmpty(suggestions)) {
            for (Suggestion<SuggestIndex> suggestion : suggestions) {
                for (CompletionSuggestOption<SuggestIndex> option : suggestion.completion().options()) {
                    String title = option.source().getTitle();
                    list.add(title);
                }
            }
        }
        return list;
    }
}
